package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Date:2025/4/9
 * Author：Ys
 * Description:
 */
object RDD_Action_1 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(sparkConf)
    //避免日志打印的过多 设置日志级别
    sc.setLogLevel("ERROR")

    //行动算子
    /*
     reduce: 将RDD中的所有元素进行聚合，聚合规则由用户自己定义
             先聚合分区内，再聚合分区间
     */
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)// 【1，2】  【3，4】
    val i: Int = rdd1.reduce( _ + _ )  // 1 + 2 =3           3 +  4 =7   3 + 7 =10
    println(i)

    /*
    collect: 将RDD中的数据拉取到Driver端内存中，形成Array
     */
    val arrs1: Array[Int] = rdd1.collect()
    println(arrs1.mkString(","))


    /*
     count:获取RDD元素的个数
     */
    val l: Long = rdd1.count()
    println(l)

    /*
     first: 返回RDD中的第一个元素
     */
    val first: Int = rdd1.first()
    println(first)

    /*
    take:返回RDD中前n个元素，组成的数组
     */
    val arrs2: Array[Int] = rdd1.take(3)
    println(arrs2.mkString(" "))

    /*
     takeOrdered:返回RDD中排序后前n个元素，组成的数组
     */
    val rdd2 = sc.makeRDD( List(2,1,4,3))
    val arrs3: Array[Int] = rdd2.takeOrdered(2)
    println(arrs3.mkString(","))

    /*
    aggregate:分区的数据通过初始值和分区内得数据进行聚合，然后再和初始值进行分区间的聚合
     */
    val res: Int = rdd1.aggregate(10)(_ + _, _ + _)
    //【1，2】  【3，4】==> 1+2+10 = 13       3+4+10 = 17         10+13+17 = 40
    println(res)

    val res2 = rdd1.fold(10)(_+_)
    println(res2)

    /*
    countByKey:统计每种key出现的次数
     */
    val rdd3 = sc.makeRDD(List(("a",1),("b",2),("a",3),("b",4),("c",5)))
    val ckRdd: collection.Map[String, Long] = rdd3.countByKey()
    println(ckRdd)

    /*
    save相关算子
     saveAsTextFile:将RDD保存到文本文件中
     saveAsSequenceFile:将RDD保存到Sequence文件中 数据格式必须是K-V类型
     saveAsObjectFile:将RDD保存到Object文件中
     */
    rdd3.saveAsTextFile("output2")
    rdd3.saveAsObjectFile("output3")
    rdd3.saveAsSequenceFile("output4")

    sc.stop()
  }

}
